#include "Msg_q.h"

static pthread_mutex_t s_msg_q_list_mutex;
static MSG_Q_T 			*s_msg_q_list;
//public
INT32	msg_q_create(HANDLE_T*	ph_msg_hdl,
					const char*	ps_name,
					SIZE_T		z_msg_zise,
					INT16		ui2_max_msg_count)
{
	MSG_Q_T *pt_msg_q;
	SIZE_T 	message_size;
	pthread_condattr_t condattr;

	//check args
	if(ps_name == NULL || ph_msg_hdl == NULL){
		return RET_INV_ARG;
	}

	message_size = sizeof(SIZE_T) + ((z_msg_zise + 3) & ~3);

	pt_msg_q = (MSG_Q_T *)calloc(1, sizeof(MSG_Q_T) - sizeof(BYTE) + message_size * (ui2_max_msg_count + 1));
	if (pt_msg_q == NULL)
	{
		return RET_NO_RESOURCE;
	}

	pt_msg_q->read = pt_msg_q->start;
	pt_msg_q->write = pt_msg_q->start;
	pt_msg_q->end = pt_msg_q->start + message_size * (ui2_max_msg_count + 1);
	pt_msg_q->message_size = message_size;
	pt_msg_q->z_max_size = z_msg_zise;
	pt_msg_q->ui2_msg_count = ui2_max_msg_count;
	pt_msg_q->i2_refcount = 1;

	strcpy(pt_msg_q->s_name, ps_name);

	VERIFY(pthread_condattr_init(&condattr) == 0);
    //todo 注释掉了一行
	//VERIFY(pthread_condattr_setclock(&condattr, CLOCK_MONOTONIC) == 0);
    VERIFY(pthread_cond_init(&pt_msg_q->read_cond, &condattr) == 0);
	VERIFY(pthread_cond_init(&pt_msg_q->write_cond, &condattr) == 0);
	VERIFY(pthread_mutex_init(&pt_msg_q->mutex, NULL) == 0);

	VERIFY(pthread_mutex_lock(&s_msg_q_list_mutex) ==0);
	if(msg_q_find_obj(ps_name) != NULL)
	{
		VERIFY(pthread_mutex_unlock(&s_msg_q_list_mutex) == 0);
		free(pt_msg_q);
		return RET_EXIST;
	}
	msg_q_list_add(pt_msg_q);
	VERIFY(pthread_mutex_unlock(&s_msg_q_list_mutex) == 0);

	*ph_msg_hdl = (HANDLE_T)(pt_msg_q);
	return RET_OK;
}
INT32	msg_q_delete(HANDLE_T	h_msg_hdl)
{
	MSG_Q_T *pt_msg_q;
	INT16	i2_refcount;

	pt_msg_q = (MSG_Q_T *)h_msg_hdl;

	VERIFY(pthread_mutex_lock(&s_msg_q_list_mutex) == 0);
	i2_refcount = --pt_msg_q->i2_refcount;
	if(i2_refcount > 0){
		VERIFY(pthread_mutex_unlock(&s_msg_q_list_mutex) == 0);
		return RET_OK;
	}
	msg_q_list_remove(pt_msg_q);
	VERIFY(pthread_mutex_unlock(&s_msg_q_list_mutex) == 0);

	VERIFY(pthread_mutex_destroy(&pt_msg_q->mutex) == 0);
	VERIFY(pthread_cond_destroy(&pt_msg_q->write_cond) == 0);
	VERIFY(pthread_cond_destroy(&pt_msg_q->read_cond) == 0);

	free(pt_msg_q);
	return RET_OK;
}
INT32	msg_q_send(HANDLE_T		h_msg_hdl,
					const VOID*	pv_msg,
					SIZE_T		z_size,
					UINT8		ui1_pri)
{
	MSG_Q_T *pt_msg_q;
	BYTE	*write;
	int 	ret;

	if(pv_msg == NULL || z_size == 0)
	{
		return RET_INV_ARG;
	}

	pt_msg_q = (MSG_Q_T *)h_msg_hdl;

	if (z_size > pt_msg_q->z_max_size)
	{
		return RET_TOO_BIG;
	}

	ret = pthread_mutex_lock(&pt_msg_q->mutex);
	if(ret != 0)
	{
		goto err;
	}
	write = pt_msg_q->write + pt_msg_q->message_size;
	if(write == pt_msg_q->end)
	{
		write = pt_msg_q->start;
	}
	if(write == pt_msg_q->read)
	{
		VERIFY(pthread_mutex_unlock(&pt_msg_q->mutex) == 0);
		return RET_TOO_MANY;
	}
	*(SIZE_T *)(pt_msg_q->write) = z_size;
	memcpy(pt_msg_q->write + 4, pv_msg, z_size);
	pt_msg_q->write = write;
	VERIFY(pthread_cond_broadcast(&pt_msg_q->write_cond) == 0);
	VERIFY(pthread_mutex_unlock(&pt_msg_q->mutex) == 0);
	return RET_OK;
err:
	switch(ret)
	{
	case EINVAL:
		return RET_INV_HANDLE;
	default:
		return RET_FAIL;
	}
}

INT32	msg_q_receive(UINT16	*pui2_index,
					   VOID		*pv_msg,
					   SIZE_T		*pz_size,
					   HANDLE_T	*ph_msg_q_hdl_list,
					   UINT16		ui2_msg_q_hdl_count,
					   MSG_Q_OPTION_T	e_option)
{
	MSG_Q_T *pt_msg_q;
	BYTE	*read;
	SIZE_T	z_size;
	int ret;

	//check args
	if(pv_msg == NULL || ph_msg_q_hdl_list == NULL || ui2_msg_q_hdl_count == 0)
	{
		return RET_INV_ARG;
	}

	if(ui2_msg_q_hdl_count != 1)
	{
		return RET_NO_SUPPORT;
	}

	pt_msg_q = (MSG_Q_T *)(ph_msg_q_hdl_list[0]);
	ret = pthread_mutex_lock(&pt_msg_q->mutex);
	if(ret != 0)
	{
		goto err;
	}

	while(pt_msg_q->read == pt_msg_q->write)
	{
		if(e_option == MSG_Q_OPTION_NOWAIT)
		{
			VERIFY(pthread_mutex_unlock(&pt_msg_q->mutex) == 0);
			return RET_NO_MSG;
		}
		VERIFY(pthread_cond_wait(&pt_msg_q->write_cond, &pt_msg_q->mutex) == 0);
	}

	read = pt_msg_q->read;
	z_size = *(SIZE_T *)read;
	if(z_size > *pz_size)
	{
		z_size = *pz_size;
	}
	memcpy(pv_msg, read + 4, z_size);
	read += pt_msg_q->message_size;
	if(read == pt_msg_q->end)
	{
		read = pt_msg_q->start;
	}
	pt_msg_q->read = read;

	VERIFY(pthread_cond_broadcast(&pt_msg_q->read_cond) == 0);
	VERIFY(pthread_mutex_unlock(&pt_msg_q->mutex) == 0);
	
	*pui2_index =0;
	*pz_size = z_size;

	return RET_OK;
err:
	switch(ret)
	{
	case EINVAL:
		return RET_INV_HANDLE;
	default:
		return RET_FAIL;
	}
}

//private
MSG_Q_T *msg_q_find_obj(const CHAR *ps_name)
{
	MSG_Q_T *pt_msg_q = s_msg_q_list;
	if(pt_msg_q == NULL)
	{
		return NULL;
	}

	do
	{
		if(strncmp(pt_msg_q->s_name, ps_name, MSG_Q_NAME_LEN) == 0)
		{
			return pt_msg_q;
		}
		pt_msg_q = pt_msg_q->next;
	}while(pt_msg_q != s_msg_q_list);

	return NULL;
}

void 	msg_q_list_add(MSG_Q_T *pt_msg_q)
{
	if(s_msg_q_list != NULL)
	{
		pt_msg_q->previous = s_msg_q_list->previous;
		pt_msg_q->next = s_msg_q_list;
		s_msg_q_list->previous->next = pt_msg_q;
		s_msg_q_list->previous = pt_msg_q;
	}
	else
	{
		s_msg_q_list = pt_msg_q->next = pt_msg_q->previous = pt_msg_q;
	}
}

void 	msg_q_list_remove(MSG_Q_T 	*pt_msg_q)
{
	if(pt_msg_q->previous == pt_msg_q)
	{
		s_msg_q_list = NULL;
	}
	else
	{
		pt_msg_q->previous->next = pt_msg_q->next;
		pt_msg_q->next->previous = pt_msg_q->previous;
		if(s_msg_q_list == pt_msg_q)
		{
			s_msg_q_list = pt_msg_q->next;
		}
	}
}